package net.bwie.realtime.jtp.dwd.logs.car.job;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

public class AdsErrReport {
    public static void main(String[] args) {
        // 1.表执行环境
        TableEnvironment tabEnv = getTableEnv();
        // 2.映射到kafka
        createInputTable(tabEnv);
//        // 3. 数据处理-select
        Table reportTable = handle(tabEnv);
////        // 4.输出表-output:映射到doris表
        createOutputTable(tabEnv);
//        // 5.保存数据-insert
        saveToSink(tabEnv,reportTable);
    }



    private static void saveToSink(TableEnvironment tabEnv, Table reportTable) {
        // 注册Table为表，给以表名称
        tabEnv.createTemporaryView("report_table", reportTable);
        // 查询,插入
        tabEnv.executeSql(
                "INSERT INTO ner_car_report.ads_err_report\n" +
                        "SELECT\n" +
                        "    err\n" +
                        "     ,date_format(dates,'yyyy-MM-dd') AS dt\n" +
                        "     ,count(vin) AS err_count\n" +
                        "FROM ner_car_report.dws_cars_err\n" +
                        "GROUP BY err,date_format(dates,'yyyy-MM-dd')"
        );
    }
    private static void createOutputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql(
                "CREATE TABLE ner_car_report.ads_err_report\n" +
                        "(\n" +
                        "    err VARCHAR(50) COMMENT '报警信息',\n" +
                        "    dt VARCHAR(50) COMMENT '分区时间',\n" +
                        "    err_count  BIGINT COMMENT '报警总次数'\n" +
                        ")WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'ner_car_report.ads_err_report',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.label-prefix' = 'err2'\n" +
                        ");"
        );
    }
    private static Table handle(TableEnvironment tabEnv) {
        Table errNum = tabEnv.sqlQuery(
                "SELECT\n" +
                        "    err\n" +
                        "     ,date_format(dates,'yyyy-MM-dd') AS dt\n" +
                        "     ,count(vin) AS err_count\n" +
                        "FROM ner_car_report.dws_cars_err\n" +
                        "GROUP BY err,date_format(dates,'yyyy-MM-dd')"
        );
        return errNum;
    }
    private static void createInputTable(TableEnvironment tabEnv) {
        tabEnv.executeSql("CREATE DATABASE IF NOT EXISTS ner_car_report");
        tabEnv.executeSql(
                "CREATE TABLE IF NOT EXISTS ner_car_report.dws_cars_err\n" +
                        "(\n" +
                        "    vin VARCHAR(50) COMMENT '汽车ID',\n" +
                        "    err VARCHAR(50) COMMENT '报警信息',\n" +
                        "    dates  TIMESTAMP(3) COMMENT '时间'\n" +
                        "\n" +
                        "    )WITH (\n" +
                        "    'connector' = 'doris',\n" +
                        "    'fenodes' = 'node102:8030',\n" +
                        "    'table.identifier' = 'ner_car_report.dws_cars_err',\n" +
                        "    'username' = 'root',\n" +
                        "    'password' = '123456',\n" +
                        "    'sink.label-prefix' = 'err1'\n" +
                        ")"
        );
    }

    private static TableEnvironment getTableEnv() {
        // 1.环境属性设置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tabEnv = TableEnvironment.create(settings);
        // 2.配置属性设置
        Configuration configuration = tabEnv.getConfig().getConfiguration();
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("execution.checkpointing.interval", "30 s");
        return tabEnv;
    }
}